package com.alarm.ecp;

import com.alarm.ecp.pojo.AlertEvent;
import com.alarm.ecp.pojo.LogInEvent;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.List;
import java.util.Map;


/**
 * 找出那些 5 秒钟内连续登录失败的账号，然后禁止用户，再次尝试登录需要等待 1 分钟
 **/
public class LoginAlarmFlink {
    private static final Logger logger = LoggerFactory.getLogger(LoginAlarmFlink.class);
    public static void main(String[] args) throws Exception {
        Pattern<LogInEvent, LogInEvent> pattern = Pattern.<LogInEvent>begin("start").where(new IterativeCondition<LogInEvent>() {
            @Override
            public boolean filter(LogInEvent value, Context<LogInEvent> ctx) throws Exception {
                return value.getIsSuccess().equals("fail");
            }
        }).next("next").where(new IterativeCondition<LogInEvent>() {
            @Override
            public boolean filter(LogInEvent value, Context<LogInEvent> ctx) throws Exception {
                return value.getIsSuccess().equals("fail");
            }
        }).within(Time.seconds(5));

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<LogInEvent> source = env.fromElements(
                new LogInEvent(1L, "fail", 1597905234000L),
                new LogInEvent(1L, "success", 1597905235000L),
                new LogInEvent(1L, "fail", 1597905237000L),
                new LogInEvent(2L, "fail", 1597905232000L),
                new LogInEvent(2L, "fail", 1597905238000L),
                new LogInEvent(2L, "fail", 1597905242000L),
                new LogInEvent(3L, "success", 1597905240000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<LogInEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getTimeStamp())
        ).keyBy((KeySelector<LogInEvent, Object>) LogInEvent::getUserId);

        PatternStream<LogInEvent> patternStream = CEP.pattern(source, pattern);
        SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<LogInEvent, AlertEvent>() {
            @Override
            public void processMatch(Map<String, List<LogInEvent>> match, Context context, Collector<AlertEvent> out) throws Exception {
                List<LogInEvent> start = match.get("start");
                List<LogInEvent> next = match.get("next");
                System.err.println("start:" + start + ",next:" + next);
                out.collect(new AlertEvent(String.valueOf(start.get(0).getUserId()), "出现连续登陆失败"));
            }
        });

        process.printToErr();
        env.execute("execute cep");
    }
}
